package com.yykj.pharmacist.server.message;

import com.google.common.reflect.TypeToken;
import com.yykj.common.util.GsonUtil;
import com.yykj.message.center.client.dto.PrescriptionMessageBodyDTO;
import com.yykj.message.center.client.dto.TDMQMessage;
import com.yykj.pharmacist.server.webScoket.PrescriptionPushSocket;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;

@Service
@Slf4j
public class PharmacistMessageService {

    //从配置文件中获取 serviceURL 接入地址、Token 密钥、Topic 全名和 Subscription 名称（均可从控制台复制）
    @Value("${tdmq.serviceUrl}")
    private String serviceUrl;
    @Value("${tdmq.token}")
    private String token;

    @Value("${tdmq.namespace}")
    private String namespace;

    private String prefix = "persistent://pulsar-pjnbmw52podb/";

    private PulsarClient client;

    private List<Consumer<byte[]>> consumerList = new ArrayList<>();

    private Consumer<byte[]> consumer;

    MessageListener<byte[]> messageListener;

    @PostConstruct
    public void init() throws Exception {
        messageListener = new MessageListener<byte[]>() {
            @Override
            public void received(Consumer<byte[]> consumer, Message<byte[]> message) {
                try {
                    if (onMessage(message)) {
                        consumer.acknowledge(message);
                    } else {
                        consumer.negativeAcknowledge(message);
                    }
                } catch (PulsarClientException e) {
                    consumer.negativeAcknowledge(message);
                    e.printStackTrace();
                }
            }
        };

        client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(AuthenticationFactory.token(token))
                .build();
        System.out.println(">> pulsar client created.");
        //创建消费者

        consumer = client.newConsumer()
                .topic(prefix + namespace + "/prescription-created")
                .subscriptionName("pharmacist-client")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(messageListener)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();

        consumer = client.newConsumer()
                .topic(prefix + namespace + "/prescription-change")
                .subscriptionName("pharmacist-client")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(messageListener)
                .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
                .subscribe();

        consumerList.add(consumer);

    }

    @PreDestroy
    public void destroy() throws Exception {
        for (Consumer<byte[]> consumer : consumerList) {
            if (consumer != null) {
                consumer.close();
            }
            if (client != null) {
                client.close();
            }
        }

    }

    /**
     * 这里写消费业务逻辑
     *
     * @param message
     * @return return true: 消息ack  return false: 消息nack
     * @throws Exception 消息nack
     */
    private boolean onMessage(Message<byte[]> message) {
        //业务逻辑,延时类操作
        String msg = new String(message.getValue());
        log.info("药师服务 -- 收到消息：{}", msg);
        try {
            TDMQMessage tdmqMessage = GsonUtil.fromJson(new String(message.getValue()),
                    new TypeToken<TDMQMessage<PrescriptionMessageBodyDTO>>(){}.getType());

            switch (tdmqMessage.getType()){
                case PRESCRIPTION_CREATED:
                    PrescriptionPushSocket.sendPrescriptionCreatedMessageToClient((PrescriptionMessageBodyDTO)tdmqMessage.getData());
                    break;
                case PRESCRIPTION_STATUS_CHANGE:
                    PrescriptionPushSocket.sendPrescriptionStatusChangeToClient((PrescriptionMessageBodyDTO)tdmqMessage.getData());
                    break;
                default:break;
            }
            log.info("药师服务 -- 消息处理 succeed：{}", msg);
            return true;
        } catch (Exception exception) {
            log.error("药师服务 -- 消息处理 failed：{}", msg);
        }
        return false;
    }

}
